package com.simple.datax.conn.impl;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/** 
 * @Title: BaseConnector.java 
 * @Package com.simple.datax.model.connector 
 * @Description: 扩展ThreadPoolTaskExecutor用于限制一个连接器只能在线程池中出现一次。<br>
 * 这个限制主要是避免一个连接器连接传输缓慢，如果同一个连接器被丢入线程池中，可能占用所有线程，导致其他连接器的任务无法得到运行。
 * &lt;Important Note:&gt; This class should be initiated by Spring ONLY ONCE.
 * @author Jin Yi 
 * @date 2013-1-20 
 * @version V1.0
 */
public class SendThreadPoolExecutor extends ThreadPoolTaskExecutor {
	private static SendThreadPoolExecutor instance;
//	private ThreadPoolExecutor threadPoolExecutor;
	private Set<String> connectorSet = new HashSet<String>();

	/**
	 * 
	 */
	private static final long serialVersionUID = -6635958602858575004L;
	
	/**
	 * Constructor should be called by Spring context for initiated this class.
	 */
	public SendThreadPoolExecutor() {
		initializeExecutor(Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());
//		threadPoolExecutor.allowCoreThreadTimeOut(true);
//		setCorePoolSize(5);
//		setMaxPoolSize(10);
		instance = this;
	}
	
	/**
	 * Singleton function should be called by code manually.
	 * @return
	 */
	public synchronized static SendThreadPoolExecutor getInstance() {
		if (instance == null) {
			instance = new SendThreadPoolExecutor();
		}
		return instance;
	}
	
	/**
	 * Check if the key is existed, if not, start the runnable.
	 * @param task
	 * @param key
	 */
	public void execute(Runnable task, String key) {
		if (!isExisted(key)) {
			execute(task);
//			logger.debug("SendThreadPoolExecutor execute()" + key);
		}
	}
	
	/**
	 * Check if the key is existed
	 * @param connector
	 * @return
	 */
	private boolean isExisted(String connector) {
		synchronized (connectorSet) {
			if (connectorSet.contains(connector)) {
				return true;
			} else {
				connectorSet.add(connector);
//				logger.debug("SendThreadPoolExecutor add " + connector);
			}
		}
		return false;
	}
	
	/**
	 * remove the key when the runnable is done.
	 * @param connector
	 */
	public void workDone(String connector) {
		synchronized (connectorSet) {
			connectorSet.remove(connector);
//			logger.debug("SendThreadPoolExecutor release " + connector);
		}
	}
}
